#include <string>
#include <iostream>
#include <stdio.h>
#include "CThread.h"
#include <pthread.h>
#include <time.h>

using namespace std;

CThreadPool::CThreadPool(int threadNum) 
{
	init_(threadNum,threadNum);
}

CThreadPool::CThreadPool(int min, int max)
{
	init_(min,max);
}

void CThreadPool::init_(int min,int max)
{
	pthread_mutex_init(&(this->m_pthreadMutex),NULL);
	pthread_cond_init(&(this->m_pthreadCond),NULL);
	this->m_max          = max;
	this->m_min          = min;
	m_thread_num         = 0;
	m_max_task			 = 100;
	m_before_run_task_cb = NULL; 
	m_after_run_task_cb = NULL; 
	m_thread_exit_cb     = NULL;
	Create(this->m_min);
}

int CThreadPool::MoveOutIdleThread(pthread_t tid)
{
    vector<pthread_t>::iterator idleIter = m_vecIdleThread.begin();
    while(idleIter != m_vecIdleThread.end())
    {
        if(tid == *idleIter)
        {
            break;
        }
        idleIter++;
    }
    if(idleIter != m_vecIdleThread.end()){
        m_vecIdleThread.erase(idleIter);
        m_thread_num--;
        if(m_thread_exit_cb != NULL)
		{
            m_thread_exit_cb(tid);
        }
        printf("thread out -- ----m_thread_num = %d\n",m_thread_num);
    } 
    return 0;
}

int CThreadPool::MoveToIdle(pthread_t tid)
{
    vector<pthread_t>::iterator busyIter = m_vecBusyThread.begin();
    while(busyIter != m_vecBusyThread.end())
    {
        if(tid == *busyIter)
        {
            break;
        }
        busyIter++;
    }
    m_vecBusyThread.erase(busyIter);
    m_vecIdleThread.push_back(tid);
    return 0;
}

int CThreadPool::MoveToBusy(pthread_t tid)
{
    vector<pthread_t>::iterator idleIter = m_vecIdleThread.begin();
    while(idleIter != m_vecIdleThread.end())
    {
        if(tid == *idleIter)
        {
            break;
        }
        idleIter++;
    }
    m_vecIdleThread.erase(idleIter);
    m_vecBusyThread.push_back(tid);
    return 0;
}

void* CThreadPool::ThreadFunc(void * p_thread_pool)
{
    pthread_t tid = pthread_self();
    pthread_setcancelstate(PTHREAD_CANCEL_ENABLE,NULL);
    pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS,NULL);
    CThreadPool * pool = (CThreadPool *)p_thread_pool;
    
    struct timespec tp;
    clock_gettime ( CLOCK_REALTIME, &tp );
    tp.tv_sec += 60;
    tp.tv_nsec = 0;
	
    bool doing = false; //正在循环执行任务,doing时，线程在忙盒子里
    //printf("THREAD:%u:tid=%u start\n",time(0),tid);
    while(1)
	{
        pthread_mutex_lock(&(pool->m_pthreadMutex));
        vector<CTask*> & taskList = pool->m_vecTaskList;
        if(taskList.empty())
		{
            if(doing)
			{
                pool->MoveToIdle(tid);
                doing = false;
            }
            //printf("THREAD:%u:tid=%u wait,tp->tv_sec=%d\n",time(0),tid,tp.tv_sec);
            pthread_cond_timedwait(&pool->m_pthreadCond,&pool->m_pthreadMutex,&tp);
            //printf("THREAD:%u:tid=%u wake up\n",time(0),tid);
        }

        int res = 0;//调用任务运行检查函数，看资源是否准备好
        CTask* task = NULL;
        if(pool->m_before_run_task_cb != NULL)
        {
            if(pool->m_before_run_task_cb(tid) != 0)
            {
                cout<<"m_before_run_task_cb != 0, count get sphinx client!!!" << endl;
                res = 1;
            }
        }
        if( res == 0 )
		{ 
			//资源准备好了
            vector<CTask*>::iterator iter = taskList.begin();
            while(iter != taskList.end())
            {
                if(!doing)
				{
                    pool->MoveToBusy(tid);
                    doing = true;
                }
                task = *iter;
                //printf("THREAD:%u:tid=%u,task=%u\n",time(0),tid,task);
                taskList.erase(iter);
                break;
            }
        }
        pthread_mutex_unlock(&(pool->m_pthreadMutex));
        if(task != NULL )
		{
            if(doing != true) printf("error ---------------------\n");
            //printf("THREAD:%u:tid=%u,task=%u\n",time(0),tid,task);
            task->Run();
            delete task;//FIXME:这行的代码位置不是很合适，现在还没有找到合适的释放任务的地方
            if(pool->m_after_run_task_cb != NULL)
            {
                pool->m_after_run_task_cb(tid);
            }
            tp.tv_sec=time(0)+60;
        }
		else
		{
            //printf("THREAD:%u:tid=%u,task is null,doing=%s\n",time(0),tid,doing?"true":"false");
            if( time(0) < tp.tv_sec)
            {
                //printf("THREAD:%u:tid=%u,time in,go on\n",time(0),tid);
                continue;
            }
			//被唤醒，但是没有任务，判断是否需要退出
            if(!doing)
			{
				//仅空闲状态下退出
                //cout<< "m_thread_num="<<pool->m_thread_num<<",m_min="<<pool->m_min<<endl;
                if(pool->m_thread_num > pool->m_min)
				{
                    pthread_mutex_lock(&(pool->m_pthreadMutex));
                    pool->MoveOutIdleThread(tid);
                    pthread_mutex_unlock(&(pool->m_pthreadMutex));
                    break;//exit thread
                }
            } 
        }
    }
    return (void*)0;
}

/**
 * 添加任务并触发等待线程工作，如果不能加入，返回-1.
 */
int CThreadPool::AddTask(CTask *task)
{
	if(m_vecTaskList.size()>m_max_task)
	{
		return -1;
	}
    pthread_mutex_lock(&m_pthreadMutex);
    this->m_vecTaskList.push_back(task);
    int need_thread_num = this->m_vecTaskList.size() - m_thread_num;
    int call_num=0;//唤醒次数，如果同时有任务和空闲线程，增加唤醒次数
    call_num = min(m_vecTaskList.size(),m_vecIdleThread.size());
    //if(need_thread_num > 0)
	//为了避免频繁的创建后又销毁线程，不能需要就创建
    if(need_thread_num > m_thread_num)
	{
        if(need_thread_num+this->m_thread_num > this->m_max)
		{
            need_thread_num = this->m_max - this->m_thread_num;
        }
        if(need_thread_num>0)
        {
            Create(need_thread_num);
        }
    }
    //printf("THREAD:%u:add task=%u,call\n",task);
    pthread_cond_signal(&m_pthreadCond);
    pthread_mutex_unlock(&m_pthreadMutex);
    return 0;
}

int CThreadPool::Create(int thread_num)
{
    static int total_thread_num=0;
    for(int i = 0; i < thread_num;i++)
    {
        pthread_t tid = 0;
        pthread_attr_t attr;
        pthread_attr_init( &attr );
        pthread_attr_setdetachstate( &attr, PTHREAD_CREATE_DETACHED );
        pthread_create(&tid,&attr,CThreadPool::ThreadFunc,this);
        m_vecIdleThread.push_back(tid);
        m_thread_num++;
    }
    total_thread_num+=thread_num;
    //printf("THREAD:thread_num = %d,m_thread_num=%d, total_thread_num = %d\n",thread_num,m_thread_num,total_thread_num);
    return 0;
}

//测试不通过，不要调用,可能是互斥锁的原因
int CThreadPool::StopAll()
{
    vector<pthread_t>::iterator iter = m_vecIdleThread.begin();
    while(iter != m_vecIdleThread.end())
    {
        pthread_cancel(*iter);
        pthread_join(*iter,NULL);
        iter++;
    }

    iter = m_vecBusyThread.begin();
    while(iter != m_vecBusyThread.end())
    {
        pthread_cancel(*iter);
        pthread_join(*iter,NULL);
        iter++;
    }
    
    pthread_mutex_destroy(&this->m_pthreadMutex);
    pthread_cond_destroy (&this->m_pthreadCond);

    return 0;
}

